package io.mqttpush.getway.http.controller;

import io.mqttpush.common.events.CommonTopics;
import io.mqttpush.getway.GetWayConstantBean;
import io.mqttpush.getway.beans.HttpPushVo;
import io.mqttpush.getway.common.CommonResponse;
import io.mqttpush.getway.http.BcMqttHandle;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpRequest;
import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.mqtt.*;
import io.netty.handler.timeout.ReadTimeoutHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 抽象的controller
 *
 * @author tianzhenjiu
 */
public abstract class Controller {

    Logger logger = LoggerFactory.getLogger(Controller.class);


    final String finishConnectUri = "/finish";
    /**
     * 标志是否初始化了channnel
     */
    static AtomicBoolean isInit = new AtomicBoolean(false);

    final GetWayConstantBean constantBean = GetWayConstantBean.instance();

    /**
     * 服务于请求
     *
     * @param requestChannel
     * @param request
     */
    public HttpResponse service(Channel requestChannel, HttpRequest request) throws Exception {

        HttpResponse response = CommonResponse.okResponse();
        response = service0(requestChannel, request, response);

        response.headers().set(CommonResponse.CONTENT_TYPE, "text/plain");
        if (response instanceof FullHttpResponse) {
            response.headers().setInt(CommonResponse.CONTENT_LENGTH, ((FullHttpResponse) response).content().readableBytes());
        }


        return response;

    }

    public abstract HttpResponse service0(Channel requestChannel, HttpRequest request, HttpResponse response) throws Exception;

    /**
     * 把报文路由给MQTT 服务
     *
     * @param requestChannel
     * @param httpPushVo
     */
    protected HttpResponse routeData(Channel requestChannel, HttpPushVo httpPushVo, HttpRequest request, HttpResponse response) {


        Map<String, Channel> bcHttpChannels = constantBean.bcHttpChannels;

        Channel channel = null;
        String identity = httpPushVo.getFromIdentify();

        if(identity==null||identity.isEmpty()){
            return CommonResponse.badResponse();
        }

        if (bcHttpChannels.containsKey(identity)) {
            channel = bcHttpChannels.get(identity);
        }


        /**
         * 如果是要关闭
         */
        if (finishConnectUri.equalsIgnoreCase(request.uri())) {

            if (channel != null) {
                bcHttpChannels.remove(identity);
                if (channel.isActive()) {
                    channel.close();
                }
            }

            return CommonResponse.byeResponse();
        }


        channel = getAndSetChannelByIdentify(httpPushVo.getUserInfo(), identity);


        if (channel == null) {
            logger.warn("竟然找不到channel");
            return response;
        }


        channel.attr(constantBean.bcHttpCallBackAttr).set(httpPushVo.getCallback());

        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, MqttQoS.AT_LEAST_ONCE,
                true, 0);


        MqttPublishVariableHeader variableHeader = new MqttPublishVariableHeader(
                CommonTopics.ONE_2ONE_CHAT_PREFIX.concat(httpPushVo.getToIdentify()), httpPushVo.hashCode());


        MqttPublishMessage mqttPublishMessage = new MqttPublishMessage(mqttFixedHeader, variableHeader,
                httpPushVo.getByteContent());

        requestChannel.attr(ControllBeans.requestIdentifyKey).set(httpPushVo.getFromIdentify());

        if (channel.hasAttr(ControllBeans.loginKey) && channel.attr(ControllBeans.loginKey).get()) {
            channel.writeAndFlush(mqttPublishMessage);
            if (logger.isDebugEnabled()) {
                logger.debug("已登录,http写入MQTT");
            }
        } else {
            ControllBeans.mqttPublishMessages.offer(mqttPublishMessage);
            if (logger.isDebugEnabled()) {
                logger.debug("未登录,放入队列" + httpPushVo.toString());
            }
        }

        if (logger.isDebugEnabled()) {
            logger.debug(httpPushVo.getFromIdentify() + "------------------->" + httpPushVo.getToIdentify() + ":"
                    + httpPushVo.getTextcontent());
        }
        return response;
    }

    /**
     * 得到设置好了的channel，否则连接并且设置它
     *
     * @param identify
     * @return
     */
    protected Channel getAndSetChannelByIdentify(String userInfo, String identify) {

        Map<String, Channel> bcHttpChannels = constantBean.bcHttpChannels;

        boolean needInitChannel = false;
        Channel channel = null;
        if (!bcHttpChannels.containsKey(identify)) {
            needInitChannel = true;
        } else {
            channel = bcHttpChannels.get(identify);
            if (channel == null || !channel.isActive()) {
                needInitChannel = true;
            } else {
                return channel;
            }
        }

        /**
         * 连接后端MQTT，如果有旧的channel就关闭它
         */
        if (needInitChannel) {

            /**
             * TODO 如果这里有多个http请求同时携带的相同的identify ，运行的快的线程返回的channel其实是一个无效的 所以这里要对identify
             * 加锁
             */

            synchronized (identify) {

                ChannelFuture channelFuture = constantBean.httpbootstrap.connect(constantBean.mqttserver,
                        constantBean.mqttport);

                Channel oldChannel = bcHttpChannels.put(identify, channel = channelFuture.channel());

                if (oldChannel != null) {
                    oldChannel.close();
                }

                channelFuture.addListener((ChannelFuture future) -> {
                    if (future.isSuccess()) {
                        loginMqtt(future.channel(), identify, userInfo, "user123456");
                    } else {
                        logger.warn("http请求连接后端MQTT失败", future.cause());
                    }

                });
                if (logger.isDebugEnabled()) {
                    logger.debug("连接后端mq");
                }
            }

        }

        return channel;

    }

    /**
     * 登录后端MQTT 这一般只有初次连接的时候才会有
     *
     * @param channel
     * @param deviceId
     * @param username
     * @param password
     */
    protected void loginMqtt(Channel channel, String deviceId, String username, String password) {

        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.CONNECT, false, MqttQoS.AT_MOST_ONCE,
                false, 0);

        MqttConnectVariableHeader variableHeader = new MqttConnectVariableHeader(MqttVersion.MQTT_3_1_1.protocolName(),
                MqttVersion.MQTT_3_1_1.protocolLevel(), true, true, false, 0, false, false, 10);

        MqttConnectMessage connectMessage = new MqttConnectMessage(mqttFixedHeader, variableHeader,
                new MqttConnectPayload(deviceId, null, null, username, password.getBytes()));
        channel.writeAndFlush(connectMessage);

        if (logger.isDebugEnabled()) {
            logger.debug("发送登录报文成功");
        }
    }

    /**
     * 初始化配置后端连接
     */
    public void initBcChannel() {

        if (isInit.get()) {
            return;
        }

        if (!isInit.compareAndSet(false, true)) {
            return;
        }

        Bootstrap bootstrap = constantBean.httpbootstrap;
        bootstrap.group(constantBean.httpgroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {

                        /**
                         *
                         * 如果半分钟没任何报文交互就拆了
                         */
                        ch.pipeline().addLast(new ReadTimeoutHandler(30000));
                        ch.pipeline().addLast(MqttEncoder.INSTANCE, new MqttDecoder());
                        ch.pipeline().addLast(new BcMqttHandle());
                    }

                });

    }

}
